package com.xp.kettlegenerate.start;

import com.xp.kettlegenerate.enu.DataBaseDriverEnum;
import org.pentaho.di.core.Const;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleFileException;
import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.exception.KettleXMLException;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.vfs.KettleVFS;
import org.pentaho.di.core.xml.XMLHandler;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;

import java.io.DataOutputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

public class Generate {

    public static void main(String[] args) throws KettleException {

        KettleEnvironment.init();

        TransMeta transMeta = new TransMeta();

        DatabaseMeta d1 = generateDataBaseMeta(DataBaseDriverEnum.DM.dbType, "68_5236_ADW", "172.16.0.68", "5236", "ADW", "123456789", "DATA_PROCESS");
        DatabaseMeta d2 = generateDataBaseMeta(DataBaseDriverEnum.DM.dbType, "68_5236_DATA_PROCESS", "172.16.0.68", "5236", "DATA_PROCESS", "123456789", "DATA_PROCESS");
        transMeta.addDatabase(0,d1);
        transMeta.addDatabase(1,d2);
        transMeta.setName("表输入到表输出");
        PluginRegistry registry = PluginRegistry.getInstance();
        AtomicInteger locationX = new AtomicInteger(100);
        AtomicInteger locationY = new AtomicInteger(100);

        String sql = "";

//        generateHalf(transMeta, registry, locationX, locationY, d2,sql);

        String xml = transMeta.getXML();

        System.out.println("saveMetaString(xml,\"kkkk\") = " + saveMetaString(xml, "C:/Users/xp/Desktop/kkk.ktr"));
    }

    public static void generateHalf(TransMeta transMeta, PluginRegistry registry,
                                    AtomicInteger locationX,
                                    AtomicInteger locationY,
                                    DatabaseMeta d2,
                                    String  sql,
                                    String  inputName,
                                    String  outputName,
                                    String  schema
                                    ) {
        // 表输入组件
        TableInputMeta tableInputMeta = createTableInputStep(transMeta.getDatabase(0), sql);

        String pluginId = registry.getPluginId(StepPluginType.class, tableInputMeta);
        StepMeta tableInputStep = generateSetp(locationX.get(), locationY.get(), inputName,
                tableInputMeta, pluginId);
        transMeta.addStep(tableInputStep);

        // 横坐标 加200
        locationX.addAndGet(200);


        TableOutputMeta tableOutputMeta = createTableOutputMeta(transMeta, tableInputStep.getName(), d2, schema,outputName,"1000");
        String outputPluginId = registry.getPluginId(StepPluginType.class, tableOutputMeta);
        StepMeta tabOutPutStep = generateSetp(locationX.get(), locationY.get(), "表"+outputName+"输出",
                tableOutputMeta, outputPluginId);
        transMeta.addStep(tabOutPutStep);

        transMeta.addTransHop(new TransHopMeta(tableInputStep, tabOutPutStep));
    }

    public static boolean saveMetaString( String transMetaString, String fname)
    {
        boolean saved = false;

        String xml = XMLHandler.getXMLHeader() + transMetaString;
        try
        {
            //write trans meta string into a tmp file fname
            DataOutputStream dos = new DataOutputStream(KettleVFS.getOutputStream(fname, false));
            dos.write(xml.getBytes(Const.XML_ENCODING));
            dos.close();
            saved = true;
        } catch( KettleFileException e )
        {
            e.printStackTrace();
        } catch( UnsupportedEncodingException e )
        {
            e.printStackTrace();
        } catch( IOException e )
        {
            e.printStackTrace();
        }
        return saved;
    }


    /******
     *  生成步骤
     * @param locationX
     * @param locationY
     * @param stepName
     * @param stepMetaInterface
     * @return
     */
    public static StepMeta generateSetp(int locationX, int locationY, String stepName, StepMetaInterface stepMetaInterface,
                                        String pluginId) {

        StepMeta stepMeta = new StepMeta(pluginId, stepName, stepMetaInterface);
        stepMeta.setLocation(locationX, locationY);
        stepMeta.setDraw(Boolean.TRUE);

        return stepMeta;
    }

    /*********
     *  创建表输入步骤
     * @param db
     * @return
     */
    public static TableInputMeta createTableInputStep(DatabaseMeta db, String sql) {

        // 新建一个表输入元数据
        TableInputMeta tableInputMeta = new TableInputMeta();
        // 设置步骤得数据库连接
        tableInputMeta.setDatabaseMeta(db);
        // 获取查询sql 语句
        tableInputMeta.setSQL(sql);

        return tableInputMeta;
    }

    /********
     *   获取上一个步骤的字段
     * @param transMeta
     * @param preStepname
     * @return
     */
    public static String[] getPrevFiedsByArray(TransMeta transMeta, String preStepname) {

        try {
            RowMetaInterface prevStepFields = transMeta.getStepFields(preStepname);
            return prevStepFields.getFieldNames();
        } catch (KettleStepException e) {
            e.printStackTrace();
            return null;
        }
    }


    /*******
     * 表输出元数据
     * @param
     * @return
     */
    public static TableOutputMeta createTableOutputMeta(TransMeta transMeta,
                                                        String preStepname,
                                                        DatabaseMeta dataBaseMeta,
                                                        String schema,
                                                        String tabName,
                                                        String commitSize) {

        //新建一个表输出
        TableOutputMeta tableOutputMeta = new TableOutputMeta();
        // 数据库连接
        tableOutputMeta.setDatabaseMeta(dataBaseMeta);
        //目标模式
        tableOutputMeta.setSchemaName(schema);
        // 是否裁剪表
        tableOutputMeta.setTruncateTable(true);

        // 避免特殊表名  特殊处理
        //目标表
        tableOutputMeta.setTableName(tabName);
        //提交记录数量
        tableOutputMeta.setCommitSize(commitSize);
        // 指定数据库字段
        tableOutputMeta.setSpecifyFields(true);
        // 使用批量插入
        tableOutputMeta.setUseBatchUpdate(true);

        // 获取建表字段哦
        try {

            String[] fieldNames = getPrevFiedsByArray(transMeta, preStepname);

            if (null != fieldNames) {

                Map<Integer, String[]> integerMap = transition4OutPut( fieldNames);

                //流字段
                tableOutputMeta.setFieldStream(integerMap.get(1));
                //表字段
                tableOutputMeta.setFieldDatabase(integerMap.get(0));
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

        return tableOutputMeta;
    }

    /*******
     * 转换表输出数据流
     * 0 表字段
     * 1 流字段
     * @return
     */
    public static Map<Integer, String[]> transition4OutPut( String[] output) {


        HashMap<Integer, String[]> map = new HashMap<>();

        String[] outPutStream;
        String[] outPutField;


        outPutStream = new String[output.length];
        outPutField = new String[output.length];

        for (int i = 0; i < output.length; i++) {
            outPutStream[i] = output[i];
            outPutField[i] = "\""+output[i] +"\"";
        }


        map.put(0, outPutField);
        map.put(1, outPutStream);


        return map;
    }

    /*******
     *  生成对应的 元数据对象
     * @param dbType 数据库类型 具体参考 #link{DatabaseDriver} 类
     * @param metaName 元数据对象名称
     * @param host ip地址
     * @param port 端口号
     * @param userName 数据库用户名
     * @param pwd 数据库密码
     * @param dataBaseName 连接的具体某一个数据库  如果是 oracle 请填写数据库实例（orcl/EX）
     * @return
     * @throws KettleXMLException
     */
    public static DatabaseMeta generateDataBaseMeta(
            String dbType,
            String metaName,
            String host,
            String port,
            String userName,
            String pwd,
            String dataBaseName) throws KettleXMLException {

        String driverClass = DataBaseDriverEnum.getDriver(dbType);
        HashMap<String, String> map = new HashMap<>(13);

        map.put("type", "Generic database");
        map.put("access", "Native");

        map.put("database", dbType);
        map.put("name", metaName);
        map.put("server", host);
        map.put("port", port);
        map.put("userName", userName);
        map.put("password", pwd);
        map.put("driverClass", driverClass);
        map.put("dataBaseName", dataBaseName);



        return new DatabaseMeta(getDatabaseStr(map));
    }

    /*******
     *  构建通用类的 xml 对象
     * @param param 参数  ip地址 端口等参数
     * @return
     */
    public static String getDatabaseStr(Map param) {
        StringBuilder xml = new StringBuilder();
        StringBuffer url = new StringBuffer();
        xml.append("<?xml version='1.0' encoding='UTF-8'?><connection>");
        xml.append("<name>").append((String) param.get("name")).append("</name>");
        xml.append("<server>").append((String) param.get("server")).append("</server>");
        xml.append("<type>").append((String) param.get("type")).append("</type>");
        xml.append("<access>").append((String) param.get("access")).append("</access>");
        xml.append("<database>").append((String) param.get("database")).append("</database>");
        xml.append("<port>").append((String) param.get("port")).append("</port>");
        xml.append("<username>").append((String) param.get("userName")).append("</username>");
        xml.append("<password>").append((String) param.get("password")).append("</password>");
        String database = (String) param.get("database");
        xml.append("<attributes><attribute><code>CUSTOM_DRIVER_CLASS</code>");
        xml.append("<attribute>"+param.get("driverClass")+"</attribute></attribute>");
        getUrl(database,url,param);
        xml.append("<attribute><code>CUSTOM_URL</code><attribute>");
        xml.append(url);
        xml.append("</attribute></attribute></attributes>");
        xml.append("</connection>");
        return xml.toString();
    }

    /******
     *  构建数据库连接 url
     * @param dbType 数据库类型
     * @param url 数据库连接 url
     * @param param 连接参数 包含ip 端口等
     */
    public static void getUrl(String dbType, StringBuffer url, Map param){

        if(DataBaseDriverEnum.DM.dbType.equalsIgnoreCase(dbType)){
            url.append("jdbc:dm://");
            url.append((String) param.get("server"));
            url.append(":");
            url.append((String) param.get("port"));
            url.append("/");
            url.append((String) param.get("dataBaseName"));
        }else if(DataBaseDriverEnum.MYSQL.dbType.equalsIgnoreCase(dbType)){
            url.append("jdbc:mysql://");
            url.append((String) param.get("server"));
            url.append(":");
            url.append((String) param.get("port"));
            url.append("/");
            url.append((String) param.get("dataBaseName"));
            url.append("?characterEncoding=UTF-8");
        }else if(DataBaseDriverEnum.ORACLE.dbType.equalsIgnoreCase(dbType)){
            url.append("jdbc:oracle:thin:@//");
            url.append((String) param.get("server"));
            url.append(":");
            url.append((String) param.get("port"));
            url.append("/");
            url.append((String) param.get("dataBaseName"));
        }else if(DataBaseDriverEnum.SQL_SERVER.dbType.equalsIgnoreCase(dbType)){
            url.append("jdbc:sqlserver://");
            url.append((String) param.get("server"));
            url.append(":");
            url.append((String) param.get("port")).append(";");
            url.append("databaseName=");
            url.append((String) param.get("dataBaseName")).append(";");
            url.append("user=").append(param.get("userName")).append(";");
            url.append("password=").append(param.get("password")).append(";");
        }else if(DataBaseDriverEnum.KING_BASE.dbType.equalsIgnoreCase(dbType)){
            url.append("jdbc:kingbase://");
            url.append((String) param.get("server"));
            url.append(":");
            url.append((String) param.get("port"));// 54322
            url.append("/");
            url.append(param.get("dataBaseName"));
        }else{
            throw new RuntimeException("暂不支持数据库类型:\t"+dbType);
        }

    }

}
